package com.nulldev.util.internal.backport.httpclient_rw;

import java.io.EOFException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;

import com.nulldev.util.internal.backport.concurrency9.concurrent.CompletableFuture;
import com.nulldev.util.internal.backport.concurrency9.concurrent.CompletionStage;
import com.nulldev.util.internal.backport.concurrency9.concurrent.Flow;
import com.nulldev.util.internal.backport.httpclient_rw.impl.common.ILogger.Level;
import com.nulldev.util.internal.backport.httpclient_rw.impl.common.Log;
import com.nulldev.util.internal.backport.httpclient_rw.impl.common.Logger;
import com.nulldev.util.internal.backport.httpclient_rw.impl.common.MinimalFuture;
import com.nulldev.util.internal.backport.httpclient_rw.impl.common.Utils;
import com.nulldev.util.internal.backport.httpclient_rw.ResponseContent.BodyParser;
import com.nulldev.util.internal.backport.httpclient_rw.ResponseContent.UnknownLengthBodyParser;

import static com.nulldev.util.internal.backport.httpclient_rw.HttpResponse.BodySubscribers.discarding;
import static com.nulldev.util.internal.backport.httpclient_rw.HttpClient.Version.HTTP_1_1;
import static com.nulldev.util.internal.backport.httpclient_rw.RedirectFilter.HTTP_NOT_MODIFIED;
import static com.nulldev.util.internal.backport.httpclient_rw.impl.common.Utils.wrapWithExtraDetail;

class Http1Response<T> {
	private volatile ResponseContent content;
	private final HttpRequestImpl request;
	private Response response;
	private final HttpConnection connection;
	private HttpHeaders headers;
	private int responseCode;
	private final Http1Exchange<T> exchange;
	private boolean return2Cache; // return connection to cache when finished
	private final HeadersReader headersReader; // used to read the headers
	private final BodyReader bodyReader; // used to read the body
	private final Http1AsyncReceiver asyncReceiver;
	private volatile EOFException eof;
	private volatile BodyParser bodyParser;
	// max number of bytes of (fixed length) body to ignore on redirect
	private final static int MAX_IGNORE = 1024;

	// Revisit: can we get rid of this?
	static enum State {
		INITIAL,
		READING_HEADERS,
		READING_BODY,
		DONE
	}

	private volatile State readProgress = State.INITIAL;

	final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);
	final static AtomicLong responseCount = new AtomicLong();
	final long id = responseCount.incrementAndGet();
	private Http1HeaderParser hd;

	Http1Response(HttpConnection conn, Http1Exchange<T> exchange, Http1AsyncReceiver asyncReceiver) {
		this.readProgress = State.INITIAL;
		this.request = exchange.request();
		this.exchange = exchange;
		this.connection = conn;
		this.asyncReceiver = asyncReceiver;
		headersReader = new HeadersReader(this::advance);
		bodyReader = new BodyReader(this::advance);

		hd = new Http1HeaderParser();
		readProgress = State.READING_HEADERS;
		headersReader.start(hd);
		asyncReceiver.subscribe(headersReader);
	}

	String dbgTag;

	private String dbgString() {
		String dbg = dbgTag;
		if (dbg == null) {
			String cdbg = connection.dbgTag;
			if (cdbg != null) {
				dbgTag = dbg = "Http1Response(id=" + id + ", " + cdbg + ")";
			} else {
				dbg = "Http1Response(id=" + id + ")";
			}
		}
		return dbg;
	}

	// The ClientRefCountTracker is used to track the state
	// of a pending operation. Altough there usually is a single
	// point where the operation starts, it may terminate at
	// different places.
	private final class ClientRefCountTracker {
		final HttpClientImpl client = connection.client();
		// state & 0x01 != 0 => acquire called
		// state & 0x02 != 0 => tryRelease called
		byte state;

		public synchronized void acquire() {
			if (state == 0) {
				// increment the reference count on the HttpClientImpl
				// to prevent the SelectorManager thread from exiting
				// until our operation is complete.
				if (debug.on())
					debug.log("Operation started: incrementing ref count for %s", client);
				client.reference();
				state = 0x01;
			} else {
				if (debug.on())
					debug.log("Operation ref count for %s is already %s", client, ((state & 0x2) == 0x2) ? "released." : "incremented!");
				assert (state & 0x01) == 0 : "reference count already incremented";
			}
		}

		public synchronized void tryRelease() {
			if (state == 0x01) {
				// decrement the reference count on the HttpClientImpl
				// to allow the SelectorManager thread to exit if no
				// other operation is pending and the facade is no
				// longer referenced.
				if (debug.on())
					debug.log("Operation finished: decrementing ref count for %s", client);
				client.unreference();
			} else if (state == 0) {
				if (debug.on())
					debug.log("Operation finished: releasing ref count for %s", client);
			} else if ((state & 0x02) == 0x02) {
				if (debug.on())
					debug.log("ref count for %s already released", client);
			}
			state |= 0x02;
		}
	}

	private volatile boolean firstTimeAround = true;

	public CompletableFuture<Response> readHeadersAsync(Executor executor) {
		if (debug.on())
			debug.log("Reading Headers: (remaining: " + asyncReceiver.remaining() + ") " + readProgress);

		if (firstTimeAround) {
			if (debug.on())
				debug.log("First time around");
			firstTimeAround = false;
		} else {
			// with expect continue we will resume reading headers + body.
			asyncReceiver.unsubscribe(bodyReader);
			bodyReader.reset();

			hd = new Http1HeaderParser();
			readProgress = State.READING_HEADERS;
			headersReader.reset();
			headersReader.start(hd);
			asyncReceiver.subscribe(headersReader);
		}

		CompletableFuture<State> cf = headersReader.completion();
		assert cf != null : "parsing not started";
		if (debug.on()) {
			debug.log("headersReader is %s", cf == null ? "not yet started" : cf.isDone() ? "already completed" : "not yet completed");
		}

		Function<State, Response> lambda = (State completed) -> {
			assert completed == State.READING_HEADERS;
			if (debug.on())
				debug.log("Reading Headers: creating Response object;" + " state is now " + readProgress);
			asyncReceiver.unsubscribe(headersReader);
			responseCode = hd.responseCode();
			headers = hd.headers();

			response = new Response(request, exchange.getExchange(), headers, connection, responseCode, HTTP_1_1);

			if (Log.headers()) {
				StringBuilder sb = new StringBuilder("RESPONSE HEADERS:\n");
				Log.dumpHeaders(sb, "    ", headers);
				Log.logHeaders(sb.toString());
			}

			return response;
		};

		if (executor != null) {
			return cf.thenApplyAsync(lambda, executor);
		} else {
			return cf.thenApply(lambda);
		}
	}

	private boolean finished;

	synchronized void completed() {
		finished = true;
	}

	synchronized boolean finished() {
		return finished;
	}

	/**
	 * Return known fixed content length or -1 if chunked, or -2 if no
	 * content-length information in which case, connection termination delimits the
	 * response body
	 */
	int fixupContentLen(int clen) {
		if (request.method().equalsIgnoreCase("HEAD") || responseCode == HTTP_NOT_MODIFIED) {
			return 0;
		}
		if (clen == -1) {
			if (headers.firstValue("Transfer-encoding").orElse("").equalsIgnoreCase("chunked")) {
				return -1;
			}
			if (responseCode == 101) {
				// this is a h2c or websocket upgrade, contentlength must be zero
				return 0;
			}
			return -2;
		}
		return clen;
	}

	/**
	 * Read up to MAX_IGNORE bytes discarding
	 */
	public CompletableFuture<Void> ignoreBody(Executor executor) {
		int clen = (int) headers.firstValueAsLong("Content-Length").orElse(-1);
		if (clen == -1 || clen > MAX_IGNORE) {
			connection.close();
			return MinimalFuture.completedFuture(null); // not treating as error
		} else {
			return readBody(discarding(), true, executor);
		}
	}

	static final Flow.Subscription NOP = new Flow.Subscription() {
		@Override
		public void request(long n) {
		}

		public void cancel() {
		}
	};

	/**
	 * The Http1AsyncReceiver ensures that all calls to the subscriber, including
	 * onSubscribe, occur sequentially. There could however be some race conditions
	 * that could happen in case of unexpected errors thrown at unexpected places,
	 * which may cause onError to be called multiple times. The Http1BodySubscriber
	 * will ensure that the user subscriber is actually completed only once - and
	 * only after it is subscribed.
	 * 
	 * @param <U> The type of response.
	 */
	final static class Http1BodySubscriber<U> implements HttpResponse.BodySubscriber<U> {
		final HttpResponse.BodySubscriber<U> userSubscriber;
		final AtomicBoolean completed = new AtomicBoolean();
		volatile Throwable withError;
		volatile boolean subscribed;

		Http1BodySubscriber(HttpResponse.BodySubscriber<U> userSubscriber) {
			this.userSubscriber = userSubscriber;
		}

		// propagate the error to the user subscriber, even if not
		// subscribed yet.
		private void propagateError(Throwable t) {
			assert t != null;
			try {
				// if unsubscribed at this point, it will not
				// get subscribed later - so do it now and
				// propagate the error
				if (subscribed == false) {
					subscribed = true;
					userSubscriber.onSubscribe(NOP);
				}
			} finally {
				// if onError throws then there is nothing to do
				// here: let the caller deal with it by logging
				// and closing the connection.
				userSubscriber.onError(t);
			}
		}

		// complete the subscriber, either normally or exceptionally
		// ensure that the subscriber is completed only once.
		private void complete(Throwable t) {
			if (completed.compareAndSet(false, true)) {
				t = withError = Utils.getCompletionCause(t);
				if (t == null) {
					assert subscribed;
					try {
						userSubscriber.onComplete();
					} catch (Throwable x) {
						// Simply propagate the error by calling
						// onError on the user subscriber, and let the
						// connection be reused since we should have received
						// and parsed all the bytes when we reach here.
						// If onError throws in turn, then we will simply
						// let that new exception flow up to the caller
						// and let it deal with it.
						// (i.e: log and close the connection)
						// Note that rethrowing here could introduce a
						// race that might cause the next send() operation to
						// fail as the connection has already been put back
						// into the cache when we reach here.
						propagateError(t = withError = Utils.getCompletionCause(x));
					}
				} else {
					propagateError(t);
				}
			}
		}

		@Override
		public CompletionStage<U> getBody() {
			return userSubscriber.getBody();
		}

		@Override
		public void onSubscribe(Flow.Subscription subscription) {
			if (!subscribed) {
				subscribed = true;
				userSubscriber.onSubscribe(subscription);
			} else {
				// could be already subscribed and completed
				// if an unexpected error occurred before the actual
				// subscription - though that's not supposed
				// happen.
				assert completed.get();
			}
		}

		@Override
		public void onNext(List<ByteBuffer> item) {
			assert !completed.get();
			userSubscriber.onNext(item);
		}

		@Override
		public void onError(Throwable throwable) {
			complete(throwable);
		}

		@Override
		public void onComplete() {
			complete(null);
		}
	}

	public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p, boolean return2Cache, Executor executor) {
		this.return2Cache = return2Cache;
		final Http1BodySubscriber<U> subscriber = new Http1BodySubscriber<>(p);

		final CompletableFuture<U> cf = new MinimalFuture<>();

		int clen0 = (int) headers.firstValueAsLong("Content-Length").orElse(-1);

		final int clen = fixupContentLen(clen0);

		// expect-continue reads headers and body twice.
		// if we reach here, we must reset the headersReader state.
		asyncReceiver.unsubscribe(headersReader);
		headersReader.reset();
		ClientRefCountTracker refCountTracker = new ClientRefCountTracker();

		// We need to keep hold on the client facade until the
		// tracker has been incremented.
		connection.client().reference();
		executor.execute(() -> {
			try {
				content = new ResponseContent(connection, clen, headers, subscriber, this::onFinished);
				if (cf.isCompletedExceptionally()) {
					// if an error occurs during subscription
					connection.close();
					return;
				}
				// increment the reference count on the HttpClientImpl
				// to prevent the SelectorManager thread from exiting until
				// the body is fully read.
				refCountTracker.acquire();
				bodyParser = content.getBodyParser((t) -> {
					try {
						if (t != null) {
							try {
								subscriber.onError(t);
							} finally {
								cf.completeExceptionally(t);
							}
						}
					} finally {
						bodyReader.onComplete(t);
						if (t != null) {
							connection.close();
						}
					}
				});
				bodyReader.start(bodyParser);
				CompletableFuture<State> bodyReaderCF = bodyReader.completion();
				asyncReceiver.subscribe(bodyReader);
				assert bodyReaderCF != null : "parsing not started";
				// Make sure to keep a reference to asyncReceiver from
				// within this
				CompletableFuture<?> trailingOp = bodyReaderCF.whenComplete((s, t) -> {
					t = Utils.getCompletionCause(t);
					try {
						if (t == null) {
							if (debug.on())
								debug.log("Finished reading body: " + s);
							assert s == State.READING_BODY;
						}
						if (t != null) {
							subscriber.onError(t);
							cf.completeExceptionally(t);
						}
					} catch (Throwable x) {
						// not supposed to happen
						asyncReceiver.onReadError(x);
					} finally {
						// we're done: release the ref count for
						// the current operation.
						refCountTracker.tryRelease();
					}
				});
				connection.addTrailingOperation(trailingOp);
			} catch (Throwable t) {
				if (debug.on())
					debug.log("Failed reading body: " + t);
				try {
					subscriber.onError(t);
					cf.completeExceptionally(t);
				} finally {
					asyncReceiver.onReadError(t);
				}
			} finally {
				connection.client().unreference();
			}
		});
		try {
			p.getBody().whenComplete((U u, Throwable t) -> {
				if (t == null)
					cf.complete(u);
				else
					cf.completeExceptionally(t);
			});
		} catch (Throwable t) {
			cf.completeExceptionally(t);
			asyncReceiver.setRetryOnError(false);
			asyncReceiver.onReadError(t);
		}

		return cf.whenComplete((s, t) -> {
			if (t != null) {
				// If an exception occurred, release the
				// ref count for the current operation, as
				// it may never be triggered otherwise
				// (BodySubscriber ofInputStream)
				// If there was no exception then the
				// ref count will be/have been released when
				// the last byte of the response is/was received
				refCountTracker.tryRelease();
			}
		});
	}

	private void onFinished() {
		asyncReceiver.clear();
		if (return2Cache) {
			Log.logTrace("Attempting to return connection to the pool: {0}", connection);
			// TODO: need to do something here?
			// connection.setAsyncCallbacks(null, null, null);

			// don't return the connection to the cache if EOF happened.
			if (debug.on())
				debug.log(connection.getConnectionFlow() + ": return to HTTP/1.1 pool");
			connection.closeOrReturnToCache(eof == null ? headers : null);
		}
	}

	HttpHeaders responseHeaders() {
		return headers;
	}

	int responseCode() {
		return responseCode;
	}

// ================ Support for plugging into Http1Receiver   =================
// ============================================================================

	// Callback: Error receiver: Consumer of Throwable.
	void onReadError(Throwable t) {
		Log.logError(t);
		Receiver<?> receiver = receiver(readProgress);
		if (t instanceof EOFException) {
			debug.log(Level.DEBUG, "onReadError: received EOF");
			eof = (EOFException) t;
		}
		CompletableFuture<?> cf = receiver == null ? null : receiver.completion();
		debug.log(Level.DEBUG, () -> "onReadError: cf is " + (cf == null ? "null" : (cf.isDone() ? "already completed" : "not yet completed")));
		if (cf != null) {
			cf.completeExceptionally(t);
		} else {
			debug.log(Level.DEBUG, "onReadError", t);
		}
		debug.log(Level.DEBUG, () -> "closing connection: cause is " + t);
		connection.close();
	}

	// ========================================================================

	private State advance(State previous) {
		assert readProgress == previous;
		switch (previous) {
			case READING_HEADERS:
				asyncReceiver.unsubscribe(headersReader);
				return readProgress = State.READING_BODY;
			case READING_BODY:
				asyncReceiver.unsubscribe(bodyReader);
				return readProgress = State.DONE;
			default:
				throw new InternalError("can't advance from " + previous);
		}
	}

	Receiver<?> receiver(State state) {
		switch (state) {
			case READING_HEADERS:
				return headersReader;
			case READING_BODY:
				return bodyReader;
			default:
				return null;
		}

	}

	static abstract class Receiver<T> implements Http1AsyncReceiver.Http1AsyncDelegate {
		abstract void start(T parser);

		abstract CompletableFuture<State> completion();

		// accepts a buffer from upstream.
		// this should be implemented as a simple call to
		// accept(ref, parser, cf)
		public abstract boolean tryAsyncReceive(ByteBuffer buffer);

		public abstract void onReadError(Throwable t);

		// handle a byte buffer received from upstream.
		// this method should set the value of Http1Response.buffer
		// to ref.get() before beginning parsing.
		abstract void handle(ByteBuffer buf, T parser, CompletableFuture<State> cf);

		// resets this objects state so that it can be reused later on
		// typically puts the reference to parser and completion to null
		abstract void reset();

		// accepts a byte buffer received from upstream
		// returns true if the buffer is fully parsed and more data can
		// be accepted, false otherwise.
		final boolean accept(ByteBuffer buf, T parser, CompletableFuture<State> cf) {
			if (cf == null || parser == null || cf.isDone())
				return false;
			handle(buf, parser, cf);
			return !cf.isDone();
		}

		public abstract void onSubscribe(AbstractSubscription s);

		public abstract AbstractSubscription subscription();

	}

	// Invoked with each new ByteBuffer when reading headers...
	final class HeadersReader extends Receiver<Http1HeaderParser> {
		final Consumer<State> onComplete;
		volatile Http1HeaderParser parser;
		volatile CompletableFuture<State> cf;
		volatile long count; // bytes parsed (for debug)
		volatile AbstractSubscription subscription;

		HeadersReader(Consumer<State> onComplete) {
			this.onComplete = onComplete;
		}

		@Override
		public AbstractSubscription subscription() {
			return subscription;
		}

		@Override
		public void onSubscribe(AbstractSubscription s) {
			this.subscription = s;
			s.request(1);
		}

		@Override
		void reset() {
			cf = null;
			parser = null;
			count = 0;
			subscription = null;
		}

		// Revisit: do we need to support restarting?
		@Override
		final void start(Http1HeaderParser hp) {
			count = 0;
			cf = new MinimalFuture<>();
			parser = hp;
		}

		@Override
		CompletableFuture<State> completion() {
			return cf;
		}

		@Override
		public final boolean tryAsyncReceive(ByteBuffer ref) {
			boolean hasDemand = subscription.demand().tryDecrement();
			assert hasDemand;
			boolean needsMore = accept(ref, parser, cf);
			if (needsMore)
				subscription.request(1);
			return needsMore;
		}

		@Override
		public final void onReadError(Throwable t) {
			t = wrapWithExtraDetail(t, parser::currentStateMessage);
			Http1Response.this.onReadError(t);
		}

		@Override
		final void handle(ByteBuffer b, Http1HeaderParser parser, CompletableFuture<State> cf) {
			assert cf != null : "parsing not started";
			assert parser != null : "no parser";
			try {
				count += b.remaining();
				if (debug.on())
					debug.log("Sending " + b.remaining() + "/" + b.capacity() + " bytes to header parser");
				if (parser.parse(b)) {
					count -= b.remaining();
					if (debug.on())
						debug.log("Parsing headers completed. bytes=" + count);
					onComplete.accept(State.READING_HEADERS);
					cf.complete(State.READING_HEADERS);
				}
			} catch (Throwable t) {
				if (debug.on())
					debug.log("Header parser failed to handle buffer: " + t);
				cf.completeExceptionally(t);
			}
		}

		@Override
		public void close(Throwable error) {
			// if there's no error nothing to do: the cf should/will
			// be completed.
			if (error != null) {
				CompletableFuture<State> cf = this.cf;
				if (cf != null) {
					if (debug.on())
						debug.log("close: completing header parser CF with " + error);
					cf.completeExceptionally(error);
				}
			}
		}
	}

	// Invoked with each new ByteBuffer when reading bodies...
	final class BodyReader extends Receiver<BodyParser> {
		final Consumer<State> onComplete;
		volatile BodyParser parser;
		volatile CompletableFuture<State> cf;
		volatile AbstractSubscription subscription;

		BodyReader(Consumer<State> onComplete) {
			this.onComplete = onComplete;
		}

		@Override
		void reset() {
			parser = null;
			cf = null;
			subscription = null;
		}

		// Revisit: do we need to support restarting?
		@Override
		final void start(BodyParser parser) {
			cf = new MinimalFuture<>();
			this.parser = parser;
		}

		@Override
		CompletableFuture<State> completion() {
			return cf;
		}

		@Override
		public final boolean tryAsyncReceive(ByteBuffer b) {
			return accept(b, parser, cf);
		}

		@Override
		public final void onReadError(Throwable t) {
			if (t instanceof EOFException && bodyParser != null && bodyParser instanceof UnknownLengthBodyParser) {
				((UnknownLengthBodyParser) bodyParser).complete();
				return;
			}
			t = wrapWithExtraDetail(t, parser::currentStateMessage);
			Http1Response.this.onReadError(t);
		}

		@Override
		public AbstractSubscription subscription() {
			return subscription;
		}

		@Override
		public void onSubscribe(AbstractSubscription s) {
			this.subscription = s;
			try {
				parser.onSubscribe(s);
			} catch (Throwable t) {
				cf.completeExceptionally(t);
				throw t;
			}
		}

		@Override
		final void handle(ByteBuffer b, BodyParser parser, CompletableFuture<State> cf) {
			assert cf != null : "parsing not started";
			assert parser != null : "no parser";
			try {
				if (debug.on())
					debug.log("Sending " + b.remaining() + "/" + b.capacity() + " bytes to body parser");
				parser.accept(b);
			} catch (Throwable t) {
				if (debug.on())
					debug.log("Body parser failed to handle buffer: " + t);
				if (!cf.isDone()) {
					cf.completeExceptionally(t);
				}
			}
		}

		final void onComplete(Throwable closedExceptionally) {
			if (cf.isDone())
				return;
			if (closedExceptionally != null) {
				cf.completeExceptionally(closedExceptionally);
			} else {
				onComplete.accept(State.READING_BODY);
				cf.complete(State.READING_BODY);
			}
		}

		@Override
		public final void close(Throwable error) {
			CompletableFuture<State> cf = this.cf;
			if (cf != null && !cf.isDone()) {
				// we want to make sure dependent actions are triggered
				// in order to make sure the client reference count
				// is decremented
				if (error != null) {
					if (debug.on())
						debug.log("close: completing body parser CF with " + error);
					cf.completeExceptionally(error);
				} else {
					if (debug.on())
						debug.log("close: completing body parser CF");
					cf.complete(State.READING_BODY);
				}
			}
		}

		@Override
		public String toString() {
			return super.toString() + "/parser=" + String.valueOf(parser);
		}
	}
}
